package com.shujia.flink.state;

import lombok.AllArgsConstructor;
import lombok.Data;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class Demo04FraudCheck {
    public static void main(String[] args) throws Exception {
        // 对某个人的交易流水进行欺诈检测：如果有一笔交易小于一元，然后紧接着的一笔交易大于500，则判断有欺诈风险
        /*
         * 1,1000
         * 1,500
         * 1,200
         * 1,0.1
         * 1,1000
         * 1,0.1
         * 1,300
         */

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> transDS = env.socketTextStream("master", 8888);
        SingleOutputStreamOperator<MyTrans> myTransDS = transDS.map(line -> {
            String[] split = line.split(",");
            return new MyTrans(split[0], Double.parseDouble(split[1]));
        });

        myTransDS
                .keyBy(MyTrans::getId)// 按照每个人的户号进行分组
                .process(new KeyedProcessFunction<String, MyTrans, String>() {

                    ValueState<Boolean> flagState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        RuntimeContext context = getRuntimeContext();
                        flagState = context.getState(new ValueStateDescriptor<Boolean>("flag", Types.BOOLEAN));
                    }

                    @Override
                    public void processElement(MyTrans value, KeyedProcessFunction<String, MyTrans, String>.Context ctx, Collector<String> out) throws Exception {
                        // 获取上一条纪录的状态，如果为true，则表示上一条记录是小于1的，则需要对当前记录进行是否大于500的判断
                        // 如果为false，则只需要判断当前记录中的金额是否小于1
                        Boolean flag = flagState.value();
                        if(flag==null){
                            flag = false;
                        }
                        // 获取每笔交易中的交易金额
                        Double trans = value.getTrans();
                        if (trans < 1) {
                            flagState.update(true);
                        }

                        if (flag) {
                            if (trans > 500) {
                                System.out.println("存在交易风险");
                            }
                            flagState.update(false);
                        }
                    }
                });
        // flink不需要action算子触发任务，由事件触发
        env.execute();
    }
}

@Data
@AllArgsConstructor
class MyTrans {
    String id;
    Double trans;
}